package com.laosg.spark.base.oracle

import java.sql.{DriverManager, ResultSet}

import org.apache.spark.sql.SparkSession

/**
  * Created by kaimin on 15/5/2019.
  * time : 13:23
  * spark写racle数据库
  */
object OracleWriteOperation {

  val driver="oracle.jdbc.driver.OracleDriver"
  val connection = "jdbc:oracle:thin:@192.168.2.110:1521:orcl"
  val dbName = "emrqc"
  val dbPwd = "emrqc"

  def main(args: Array[String]): Unit = {

   /* val conf = new SparkConf().setAppName("oracle")
    val sc = new SparkContext(conf)
    val data = new JdbcRDD(sc,createConnection,"select * from sqoop_test",lowerBound = 1,upperBound = 2,numPartitions = 1,mapRow = extractResult)

    println(data.collect().toList)

    sc.stop()*/
    val sparkSession = SparkSession.builder()
        .appName("oracle_test")
      .getOrCreate()
    val jdbcConf = sparkSession.read
      .format("jdbc")
        .option("dbtable","(select * from SQOOP_TEST) t")
      .option("url",connection)
      .option("user",dbName)
      .option("password",dbPwd).load()


    println(jdbcConf)
    println("----------------")
    jdbcConf.createOrReplaceTempView("SQOOP_TEST")
    sparkSession.sql("insert into SQOOP_TEST values(3,\"通过spark插入数据\",19)").show()
    sparkSession.stop()
  }

  /**
    * 获取数据库连接
    * @return
    */
  def createConnection()={
    Class.forName(driver).newInstance()
    DriverManager.getConnection(connection,dbName,dbPwd)
  }

  //处理结果集
  def extractResult(r:ResultSet)={
    (r.getString(1),r.getString(2),r.getString(3))
  }






}
